import sqlite3
from sqlalchemy import pool
from typing import Optional, Any, cast
from sqlalchemy.engine.interfaces import DBAPIConnection


class SQLitePool:
    __instance = None
    _connection_pool: Optional[pool.QueuePool] = None
    _db_path: str  # SQLite 数据库文件路径

    def __new__(cls, db_path: str = "default.db", *args, **kwargs):
        # 单例模式：确保全局唯一实例（即使传入不同 db_path，也只保留首次初始化的路径）
        if cls.__instance is None:
            cls.__instance = super().__new__(cls, *args, **kwargs)
            cls.__instance._db_path = db_path
            cls.__instance._initialize_pool()
        return cls.__instance

    def __init__(self, db_path: str = "default.db") -> None:
        self.connect: Optional[Any] = None  # 当前连接
        self.cursor: Optional[Any] = None   # 当前游标

    def _initialize_pool(self):
        """初始化 SQLAlchemy 连接池"""
        self._connection_pool = pool.QueuePool(
            creator=lambda: self._create_sqlite_connection(),  # 连接创建函数
            pool_size=3,          # 核心连接数（SQLite 不宜过大）
            max_overflow=2,       # 最大临时连接数（总连接数 = pool_size + max_overflow）
            timeout=30,           # 获取连接超时时间（秒）
            recycle=3600,         # 连接回收时间（避免长时间闲置）
            echo=False            # 关闭 SQL 日志（调试时可设为 True）
        )

    def _create_sqlite_connection(self) -> DBAPIConnection:
        """创建 SQLite 连接（适配 SQLAlchemy 接口）"""
        conn = sqlite3.connect(
            database=self._db_path,
            check_same_thread=False,  # 允许跨线程使用连接（需自行保证线程安全）
            isolation_level=None      # 禁用自动事务（如需事务需手动控制）
        )
        # 设置游标返回字典格式（可选，方便使用）
        conn.row_factory = sqlite3.Row
        return cast(DBAPIConnection, conn)

    def __enter__(self) -> "SQLitePool":
        """上下文管理器：获取连接和游标"""
        if self._connection_pool is None:
            raise RuntimeError("连接池未初始化")
        self.connect = self._connection_pool.connect()  # 从池获取连接
        self.cursor = self.connect.cursor()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """上下文管理器：释放连接（归还池）"""
        if self.cursor:
            self.cursor.close()
        if self.connect:
            self.connect.close()  # 归还连接到池（非真正关闭文件）
        self.cursor = None
        self.connect = None

    # 封装常用数据库操作
    def execute(self, query: str, params: Optional[tuple] = None) -> int:
        if not self.cursor:
            raise RuntimeError("请在 with 语句中使用")
        return self.cursor.execute(query, params or ())

    def select_one(self, query: str, params: Optional[tuple] = None) -> Optional[dict]:
        if not self.cursor:
            raise RuntimeError("请在 with 语句中使用")
        # 强制转换为非空 Cursor 类型
        cursor = cast(sqlite3.Cursor, self.cursor)
        self.execute(query, params)
        row = cursor.fetchone()  # 无警告
        return dict(row) if row else None

    def select_all(self, query: str, params: Optional[tuple] = None) -> list[dict]:
        if not self.cursor:
            raise RuntimeError("请在 with 语句中使用")
        # 强制转换为非空 Cursor 类型
        cursor = cast(sqlite3.Cursor, self.cursor)
        self.execute(query, params)
        rows = cursor.fetchall()
        return [dict(row) for row in rows]

    def insert(self, query: str, params: Optional[tuple] = None) -> int:
        self.execute(query, params)
        return self.cursor.lastrowid  # type: ignore # 返回自增 ID（若表有自增主键）

    def update(self, query: str, params: Optional[tuple] = None) -> int:
        self.execute(query, params)
        return self.cursor.rowcount  # type: ignore # 返回受影响行数

    def delete(self, query: str, params: Optional[tuple] = None) -> int:
        self.execute(query, params)
        return self.cursor.rowcount  # type: ignore


if __name__ == "__main__":
    with SQLitePool("default.sqlite3") as conn:
        conn.execute(
            "CREATE TABLE IF NOT EXISTS users (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT)")
        conn.insert("INSERT INTO users (name) VALUES ('John')")
        print(conn.select_all("SELECT id, name FROM users"))
